//package com.example.service;
//
//import com.example.Utils.Env;
//import com.example.Utils.StockTradeMapper;
//import com.example.entity.StockTrade;
//import jakarta.annotation.PostConstruct;
//import jakarta.annotation.Resource;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.api.common.functions.AggregateFunction;
//import org.apache.flink.api.common.functions.MapFunction;
//import org.apache.flink.api.common.serialization.SimpleStringSchema;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.connector.kafka.source.KafkaSource;
//import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
//import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
//import org.apache.flink.streaming.api.windowing.time.Time;
//import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
//import org.springframework.stereotype.Component;
//import redis.clients.jedis.Jedis;
//
//@Component
//@Slf4j
//public class TradeCal {
//    @Resource
//    private Env env_;
//
//    @PostConstruct
//    public void init() {
//        new Thread(() -> {
//            try {
//                startFlinkJob();
//            } catch (Exception e) {
//                log.error("Error starting Flink job", e);
//            }
//        }).start();
//    }
//
//    public  void startFlinkJob() throws Exception {
//        // 创建 Kafka source
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        env.setParallelism(1);
//
//        // 创建 Kafka source
//        KafkaSource<String> source = KafkaSource.<String>builder()
//                .setBootstrapServers("192.168.88.135:19092,192.168.88.135:29092,192.168.88.135:39092")
//                .setTopics("stock-trades-")
//                .setGroupId("stock-analysis-group")
//                .setStartingOffsets(OffsetsInitializer.earliest())
//                .setValueOnlyDeserializer(new SimpleStringSchema())
//                .build();
//        log.info("数据源是" + source);
//
//        // 创建数据流并添加水印(以计算机处理速度)
//        DataStream<StockTrade> tradeStream = env
//                .fromSource(
//                        source,
//                        WatermarkStrategy.noWatermarks(), // 禁用水印
//                        "Kafka Source"
//                )
//                .map(new StockTradeMapper());
//
//        log.info("数据源搞好了");
//
//        // 1 秒窗口
//        tradeStream
//                .map(new MapFunction<StockTrade, Long>() {
//                    @Override
//                    public Long map(StockTrade value) throws Exception {
//                        return 1L; // 每个元素计数为 1
//                    }
//                })
//                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(1))) // 使用处理时间窗口
//                .aggregate(new ProcessingSpeedAggregator())
//                .addSink(new ProcessingSpeedRedisSink(redisConfig));
//
//        System.out.println("Processing speed analysis started");
//        env.execute("Processing Speed Analysis");
//    }
//
//    public FlinkJedisPoolConfig redisConfig = new FlinkJedisPoolConfig.Builder()
//            .setHost("localhost")
//            .setPort(6379)
//            .build();
//
//public static class ProcessingSpeedRedisSink extends RichSinkFunction<Long> {
//    private transient Jedis jedis;
//    private final FlinkJedisPoolConfig config;
//
//    public ProcessingSpeedRedisSink(FlinkJedisPoolConfig config) {
//        this.config = config;
//    }
//
//    @Override
//    public void open(Configuration parameters) throws Exception {
//        super.open(parameters);
//        jedis = new Jedis(config.getHost(), config.getPort());
//    }
//
//    @Override
//    public void close() throws Exception {
//        if (jedis != null) {
//            jedis.close();
//        }
//        super.close();
//    }
//
//    @Override
//    public void invoke(Long value, Context context) throws Exception {
//
//
//        // 使用时间戳作为键的一部分，确保每个键都是唯一的
//        long timestamp = context.currentProcessingTime();
//        String key = "processing_speed_"; // 存储处理速度的键
//
//        // 直接将当前窗口的计数值设置到 Redis 中
//        jedis.set(key, String.valueOf(value));
//    }
//}
//
//
//    // 定义聚合函数
//    public static class ProcessingSpeedAggregator implements AggregateFunction<Long, Long, Long> {
//
//        @Override
//        public Long createAccumulator() {
//            return 0L;
//        }
//
//        @Override
//        public Long add(Long value, Long accumulator) {
//            return accumulator + value; // 每次调用 add 方法时，计数加 1
//        }
//
//        @Override
//        public Long getResult(Long accumulator) {
//            return accumulator;
//        }
//
//        @Override
//        public Long merge(Long a, Long b) {
//            return a + b; // 合并两个计数
//        }
//    }
//}
